#!/usr/bin/env python3

import shutil
import sys
import os
import os.path
import signal
import re
import json
import copy
import traceback

from argparse import ArgumentParser
import shlex
import subprocess
from subprocess import Popen
from subprocess import PIPE
from subprocess import CalledProcessError
from subprocess import TimeoutExpired
from datetime import datetime
from time import time, sleep
from errno import ESRCH
try:
    import termcolor
except ImportError:
    termcolor = None
import random
import string
import multiprocessing
from contextlib import closing

DISTRIBUTED_DDL_TIMEOUT_MSG = "is executing longer than distributed_ddl_task_timeout"

MESSAGES_TO_RETRY = [
    "DB::Exception: ZooKeeper session has been expired",
    "DB::Exception: Connection loss",
    "Coordination::Exception: Session expired",
    "Coordination::Exception: Connection loss",
    "Coordination::Exception: Operation timeout",
    "DB::Exception: Operation timeout",
    "Operation timed out",
    "ConnectionPoolWithFailover: Connection failed at try",
    "DB::Exception: New table appeared in database being dropped or detached. Try again",
    "is already started to be removing by another replica right now",
    "DB::Exception: No more packets are available",
    DISTRIBUTED_DDL_TIMEOUT_MSG # FIXME
]

MAX_RETRIES = 4

class Terminated(KeyboardInterrupt):
    pass

def signal_handler(sig, frame):
    raise Terminated(f'Terminated with {sig} signal')

def stop_tests():
    global stop_tests_triggered_lock
    global stop_tests_triggered

    with stop_tests_triggered_lock:
        if not stop_tests_triggered.is_set():
            stop_tests_triggered.set()

            # send signal to all processes in group to avoid hung check triggering
            # (to avoid terminating clickhouse-test itself, the signal should be ignored)
            signal.signal(signal.SIGTERM, signal.SIG_IGN)
            os.killpg(os.getpgid(os.getpid()), signal.SIGTERM)
            signal.signal(signal.SIGTERM, signal.SIG_DFL)

def collect_new_asan_log_after_test(sanitizer_log_dir=None, tmp_sanitizer_log=None, case_name='test_00001'):

    if "ASAN_LOG_PATH" in os.environ and sanitizer_log_dir == None:
        sanitizer_log_dir = os.environ['ASAN_LOG_PATH']
        tmp_sanitizer_log = os.path.join(sanitizer_log_dir,'tmp')

    individual_sanitizer_log_path = os.path.join(sanitizer_log_dir, 'tmp', 'individual_sanitizer_log', )
    def get_difference_in_two_list(list_a, list_b):
        difference = []
        for item in list_a:
            if item not in list_b:
                difference.append(item)
        return difference

    def get_file_names_in_dir(sanitizer_log_dir, prefix='asan.log'):
        sanitizer_file_names = set()

        for root, dirs, files in os.walk(sanitizer_log_dir):
            for file in files:
                if file.startswith(prefix):
                    sanitizer_file_names.add(file)
        return sorted(sanitizer_file_names)

    def clean_asan_log_in_dir(root):
        file_names = os.listdir(root)
        for file_name in file_names:
            if os.path.isdir(os.path.join(root,file_name)):
                continue
            delimiter_line_number, indicator_line_number, log_data = get_line_number(root, file_name)
            result = get_splitted_log(delimiter_line_number, indicator_line_number, log_data)
            if result is None:
                os.remove(os.path.join(root,file_name))
            else:
                with open(os.path.join(root, file_name), 'w+') as cleaned_log:
                    for line in result:
                        cleaned_log.writelines(line)

    clean_asan_log_in_dir(sanitizer_log_dir)

    sanitizer_log_names = get_file_names_in_dir(sanitizer_log_dir)
    tmp_sanitizer_log_names = get_file_names_in_dir(tmp_sanitizer_log)
    new_log_files = get_difference_in_two_list(sanitizer_log_names, tmp_sanitizer_log_names)

    if new_log_files != []:  # if new asan log generated
        new_log_files = get_difference_in_two_list(sanitizer_log_names, tmp_sanitizer_log_names)

        print(f"\nWarnning! new senitizer log generated after case {case_name} - {new_log_files}")
        consolidated_log_tmp = []
        with open(os.path.join(individual_sanitizer_log_path, case_name), 'w+') as consolidated_log: # consolidate logs to cases
            for new_log_file in new_log_files:
                log_file_path = os.path.join(sanitizer_log_dir, new_log_file)
                tmp_file_path = os.path.join(tmp_sanitizer_log, new_log_file)
                with open(log_file_path, 'r') as new_log_files:
                    consolidated_log_tmp.append(f"\nlog {new_log_file}")
                    consolidated_log_tmp.append(new_log_files.read())
                    print('\n'.join(consolidated_log_tmp))
                os.system(f" cat '{log_file_path}' > '{tmp_file_path}'")
            consolidated_log.write('\n'.join(consolidated_log_tmp))

    # if asan log name are same, then diff each file， save diff result to tmp_sanitizer_log_for_cases
    for file_name in sanitizer_log_names:
        log_file_path = os.path.join(sanitizer_log_dir, file_name)
        tmp_file_path = os.path.join(tmp_sanitizer_log, file_name)
        diff_result = os.popen(f"diff -u -w --strip-trailing-cr '{tmp_file_path}' '{log_file_path}'").read()
        if len(diff_result) != 0:
            print(f"\nWarnning! new content appended after case {case_name} - {file_name}")
            print(diff_result)
            os.system(f" cat '{log_file_path}' > '{tmp_file_path}'")
            with open(os.path.join(individual_sanitizer_log_path,case_name), 'a+') as f:
                f.write(diff_result)

# check indicator in sanitizer log:
def is_sanitizer_pass():
    if "ASAN_LOG_PATH" in os.environ:
        ASAN_LOG_PATH = os.environ['ASAN_LOG_PATH']
        print(f'\nASAN enbaled, the ASAN_LOG_PATH is:{ASAN_LOG_PATH}')
        return check_asan_log( os.environ['ASAN_LOG_PATH'])
    else:
        print('\nis_sanitizer_pass skipped due to ASAN_OPTIONS not in environ')
        return True

def is_block_has_indicator(start,end,indicator_line_number):
    for line_number in indicator_line_number:
        if end >= line_number and start <= line_number:
            return True
    return False

def get_line_number(log_dir,log_file_name):
    indicator = 'StackTrace::StackTrace()'
    delimiter_line = '=================================================================\n'

    with open(os.path.join(log_dir, log_file_name), "r",errors='replace') as log:
        log_data = log.readlines()
        delimiter_line_number = []
        indicator_line_number = []
        for line_count, line in enumerate(log_data):
            if line == delimiter_line:
                delimiter_line_number.append(line_count)
            elif indicator in line:
                indicator_line_number.append(line_count)
    return delimiter_line_number, indicator_line_number, log_data

def get_splitted_log(delimiter_line_number, indicator_line_number, log_data):
    result = []
    len_delimiter_line_number = len(delimiter_line_number)

    if len_delimiter_line_number == 0:
        return None
    elif len_delimiter_line_number == 1:
        start = delimiter_line_number[0]
        end = len(log_data)
        if not is_block_has_indicator(start, end, indicator_line_number):
            result += log_data[start:end]
            return result
        else:
            return None
    else:
        total_block = len_delimiter_line_number -1
        for i in range(0,total_block):
            start = delimiter_line_number[i]
            end = delimiter_line_number[i+1]
            if not is_block_has_indicator(start,end,indicator_line_number):
                result += log_data[start:end]
        start = delimiter_line_number[-1]
        end = len(log_data)
        if not is_block_has_indicator(start, end, indicator_line_number):
            result += log_data[start:end]
        return result

def check_asan_log(log_dir):
    result_is_pass = True
    log_file_names = []

    for log_name in os.listdir(log_dir):
        path = os.path.join(log_dir, log_name)
        if os.path.isfile(path):
            log_file_names.append(log_name)

    log_file_names = [x for x in log_file_names if x.startswith('asan.log')]

    if log_file_names != []: print(f'some log files have been found in {log_dir} :{log_file_names}')
    for log_file_name in log_file_names:
        delimiter_line_number, indicator_line_number, log_data = get_line_number(log_dir, log_file_name)
        result = get_splitted_log(delimiter_line_number, indicator_line_number, log_data)
        if result:
            with open(os.path.join(log_dir, 'asan_report'), "w") as f:
                result_is_pass = False
                print(f"asan log check failed, please review the asan error in {log_file_name} !")
                f.write(f"asan error foud in {log_file_name} \n")
                for item in result:
                    for item_s in item:
                            f.write(item_s)
    return result_is_pass



def json_minify(string):
    """
    Removes all js-style comments from json string. Allows to have comments in skip_list.json.
    The code taken from https://github.com/getify/JSON.minify/tree/python under the MIT license.
    """

    tokenizer = re.compile(r'"|(/\*)|(\*/)|(//)|\n|\r')
    end_slashes_re = re.compile(r'(\\)*$')

    in_string = False
    in_multi = False
    in_single = False

    new_str = []
    index = 0

    for match in re.finditer(tokenizer, string):
        if not (in_multi or in_single):
            tmp = string[index:match.start()]
            new_str.append(tmp)
        else:
            # Replace comments with white space so that the JSON parser reports
            # the correct column numbers on parsing errors.
            new_str.append(' ' * (match.start() - index))

        index = match.end()
        val = match.group()

        if val == '"' and not (in_multi or in_single):
            escaped = end_slashes_re.search(string, 0, match.start())

            # start of string or unescaped quote character to end string
            if not in_string or (escaped is None or len(escaped.group()) % 2 == 0):  # noqa
                in_string = not in_string
            index -= 1  # include " character in next catch
        elif not (in_string or in_multi or in_single):
            if val == '/*':
                in_multi = True
            elif val == '//':
                in_single = True
        elif val == '*/' and in_multi and not (in_string or in_single):
            in_multi = False
            new_str.append(' ' * len(val))
        elif val in '\r\n' and not (in_multi or in_string) and in_single:
            in_single = False
        elif not in_multi or in_single:  # noqa
            new_str.append(val)

        if val in '\r\n':
            new_str.append(val)
        elif in_multi or in_single:
            new_str.append(' ' * len(val))

    new_str.append(string[index:])
    return ''.join(new_str)

def remove_control_characters(s):
    """
    https://github.com/html5lib/html5lib-python/issues/96#issuecomment-43438438
    """
    def str_to_int(s, default, base=10):
        if int(s, base) < 0x10000:
            return chr(int(s, base))
        return default
    s = re.sub(r"&#(\d+);?", lambda c: str_to_int(c.group(1), c.group(0)), s)
    s = re.sub(r"&#[xX]([0-9a-fA-F]+);?", lambda c: str_to_int(c.group(1), c.group(0), base=16), s)
    s = re.sub(r"[\x00-\x08\x0b\x0e-\x1f\x7f]", "", s)
    return s

def get_db_engine(args, database_name):
    if args.replicated_database:
        return " ON CLUSTER test_cluster_database_replicated ENGINE=Replicated('/test/clickhouse/db/{}', '{{shard}}', '{{replica}}')".format(database_name)
    if args.db_engine:
        return " ENGINE=" + args.db_engine
    return ""   # Will use default engine

def configure_testcase_args(args, case_file, suite_tmp_dir, stderr_file):
    testcase_args = copy.deepcopy(args)

    testcase_args.testcase_start_time = datetime.now()
    testcase_basename = os.path.basename(case_file)
    testcase_args.testcase_client = f"{testcase_args.client} --log_comment='{testcase_basename}'"
    testcase_args.testcase_basename = testcase_basename

    if testcase_args.database:
        database = testcase_args.database
        os.environ.setdefault("CLICKHOUSE_DATABASE", database)
        os.environ.setdefault("CLICKHOUSE_TMP", suite_tmp_dir)

    else:
        # If --database is not specified, we will create temporary database with unique name
        # And we will recreate and drop it for each test
        def random_str(length=6):
            alphabet = string.ascii_lowercase + string.digits
            return ''.join(random.choice(alphabet) for _ in range(length))
        database = 'test_{suffix}'.format(suffix=random_str())

        if args.tenant_id is None or len(args.tenant_id) == 0:
            create_database = database
        else:
            create_database = '{prefix}.{suffix}'.format(suffix=database, prefix=args.tenant_id)

        if args.tenant_id:
            with open(stderr_file, 'w') as stderr:
                clickhouse_proc_create = Popen(shlex.split(testcase_args.testcase_client), stdin=PIPE, stdout=PIPE, stderr=stderr, universal_newlines=True)
                try:
                    clickhouse_proc_create.communicate(("CREATE DATABASE `" + create_database + "`" + get_db_engine(testcase_args, database)), timeout=testcase_args.timeout)
                except TimeoutExpired:
                    total_time = (datetime.now() - testcase_args.testcase_start_time).total_seconds()
                    return clickhouse_proc_create, "", "Timeout creating database {} before test".format(database), total_time
        else:
            with open(stderr_file, 'w') as stderr:
                clickhouse_proc_create = Popen(shlex.split(testcase_args.testcase_client), stdin=PIPE, stdout=PIPE, stderr=stderr, universal_newlines=True)
                try:
                    clickhouse_proc_create.communicate(("CREATE DATABASE " + database + get_db_engine(testcase_args, database)), timeout=testcase_args.timeout)
                except TimeoutExpired:
                    total_time = (datetime.now() - testcase_args.testcase_start_time).total_seconds()
                    return clickhouse_proc_create, "", "Timeout creating database {} before test".format(database), total_time

        os.environ["CLICKHOUSE_DATABASE"] = database
        # Set temporary directory to match the randomly generated database,
        # because .sh tests also use it for temporary files and we want to avoid
        # collisions.
        testcase_args.test_tmp_dir = os.path.join(suite_tmp_dir, database)
        os.mkdir(testcase_args.test_tmp_dir)
        os.environ.setdefault("CLICKHOUSE_TMP", testcase_args.test_tmp_dir)

    testcase_args.testcase_database = database

    return testcase_args

def run_single_test(args, ext, server_logs_level, client_options, case_file, stdout_file, stderr_file):
    client = args.testcase_client
    start_time = args.testcase_start_time
    database = args.testcase_database

    #replace tenant_id format from a.b to a`b
    if args.tenant_id is None or len(args.tenant_id) == 0:
        drop_database = args.testcase_database
        default_database = drop_database
    else:
        drop_database = '{prefix}.{suffix}'.format(suffix=args.testcase_database, prefix=args.tenant_id)
        default_database = '{prefix}\`{suffix}'.format(suffix=args.testcase_database, prefix=args.tenant_id)

    # This is for .sh tests
    os.environ["CLICKHOUSE_LOG_COMMENT"] = args.testcase_basename

    params = {
        'client': client + ' --database="' + default_database + '"',
        'logs_level': server_logs_level,
        'options': client_options,
        'test': case_file,
        'stdout': stdout_file,
        'stderr': stderr_file,
    }

    pattern = "{test} > {stdout} 2> {stderr}"


    if ext == '.sql':
        pattern = "{client} --send_logs_level={logs_level} --testmode --multiquery {options} < " + pattern
    command = pattern.format(**params).replace("$$", "\\$\\$")

    proc = Popen(command, shell=True, env=os.environ)


    while (datetime.now() - start_time).total_seconds() < args.timeout and proc.poll() is None:
        sleep(0.01)

    need_drop_database = not args.database
    if need_drop_database and args.no_drop_if_fail:
        maybe_passed = (proc.returncode == 0) and (proc.stderr is None) and (proc.stdout is None or 'Exception' not in proc.stdout)
        need_drop_database = not maybe_passed

    if need_drop_database:
        if args.tenant_id:
            with open(stderr_file, 'a') as stderr:
                clickhouse_proc_create = Popen(shlex.split(client), stdin=PIPE, stdout=PIPE, stderr=stderr, universal_newlines=True)
            seconds_left = max(args.timeout - (datetime.now() - start_time).total_seconds(), 20)
            try:
                drop_database_query = "DROP DATABASE `" + drop_database + "`"
                if args.replicated_database:
                    drop_database_query += " ON CLUSTER test_cluster_database_replicated"
                clickhouse_proc_create.communicate((drop_database_query), timeout=seconds_left)
            except TimeoutExpired:
                # kill test process because it can also hung
                if proc.returncode is None:
                    try:
                        proc.kill()
                    except OSError as e:
                        if e.errno != ESRCH:
                            raise

                total_time = (datetime.now() - start_time).total_seconds()
                return clickhouse_proc_create, "", "Timeout dropping database {} after test".format(database), total_time
        else:
            with open(stderr_file, 'a') as stderr:
                clickhouse_proc_create = Popen(shlex.split(client), stdin=PIPE, stdout=PIPE, stderr=stderr, universal_newlines=True)
            seconds_left = max(args.timeout - (datetime.now() - start_time).total_seconds(), 20)
            try:
                drop_database_query = "DROP DATABASE " + database
                if args.replicated_database:
                    drop_database_query += " ON CLUSTER test_cluster_database_replicated"
                clickhouse_proc_create.communicate((drop_database_query), timeout=seconds_left)
            except TimeoutExpired:
                # kill test process because it can also hung
                if proc.returncode is None:
                    try:
                        proc.kill()
                    except OSError as e:
                        if e.errno != ESRCH:
                            raise

                total_time = (datetime.now() - start_time).total_seconds()
                return clickhouse_proc_create, "", "Timeout dropping database {} after test".format(database), total_time

        shutil.rmtree(args.test_tmp_dir)

    total_time = (datetime.now() - start_time).total_seconds()

    # Normalize randomized database names in stdout, stderr files.
    os.system("LC_ALL=C sed -i -e 's/{test_db}/default/g' {file}".format(test_db=database, file=stdout_file))
    if args.hide_db_name:
        os.system("LC_ALL=C sed -i -e 's/{test_db}/default/g' {file}".format(test_db=database, file=stderr_file))
    if args.replicated_database:
        os.system("LC_ALL=C sed -i -e 's|/auto_{{shard}}||g' {file}".format(file=stdout_file))
        os.system("LC_ALL=C sed -i -e 's|auto_{{replica}}||g' {file}".format(file=stdout_file))

    stdout = open(stdout_file, 'rb').read() if os.path.exists(stdout_file) else b''
    stdout = str(stdout, errors='replace', encoding='utf-8')
    # stderr = open(stderr_file, 'rb').read() if os.path.exists(stderr_file) else b''
    # stderr = str(stderr, errors='replace', encoding='utf-8')
    stderr_init = open(stderr_file, 'rb').readlines() if os.path.exists(stderr_file) else b''
    filtered_log_list = [log for log in stderr_init if b"<Warning>" not in log]
    stderr = ''.join(str(filtered_log, errors='replace', encoding='utf-8') for filtered_log in filtered_log_list)
    for log in stderr_init:
        if b"<Warning>" in log:
            print(f'{case_file}-----have warning!')
            print(f'{case_file}-----init_log:', ''.join(str(temp_err_init, errors='replace', encoding='utf-8') for temp_err_init in stderr_init))
            # print(f'{case_file}-----after_log:', stderr)
            break

    return proc, stdout, stderr, total_time


def need_retry(stderr):
    if stderr and stderr != '\n':
        return True
    else:
        return False
    # return any(msg in stderr for msg in MESSAGES_TO_RETRY)
    # return stderr != "" # retry for all error cases; TODO: fix me

def get_processlist(args):
    try:
        query = b"SHOW PROCESSLIST FORMAT Vertical"
        if args.replicated_database:
            query = b"SELECT materialize((hostName(), tcpPort())) as host, * " \
                    b"FROM clusterAllReplicas('test_cluster_database_replicated', system.processes) " \
                    b"WHERE query NOT LIKE '%system.processes%' FORMAT Vertical"
        clickhouse_proc = Popen(shlex.split(args.client), stdin=PIPE, stdout=PIPE, stderr=PIPE)
        (stdout, _) = clickhouse_proc.communicate((query), timeout=20)
        return False, stdout.decode('utf-8')
    except Exception as ex:
        print("Exception", ex)
        return True, ""


# collect server stacktraces using gdb
def get_stacktraces_from_gdb(server_pid):
    try:
        cmd = "gdb -batch -ex 'thread apply all backtrace' -p {}".format(server_pid)
        return subprocess.check_output(cmd, shell=True).decode('utf-8')
    except Exception as ex:
        print("Error occured while receiving stack traces from gdb: {}".format(str(ex)))
        return None


# collect server stacktraces from system.stack_trace table
# it does not work in Sandbox
def get_stacktraces_from_clickhouse(client, replicated_database=False):
    try:
        if replicated_database:
            return subprocess.check_output("{} --allow_introspection_functions=1 --skip_unavailable_shards=1 --query "
                "\"SELECT materialize((hostName(), tcpPort())) as host, thread_id, "
                "arrayStringConcat(arrayMap(x, y -> concat(x, ': ', y), arrayMap(x -> addressToLine(x), trace), "
                "arrayMap(x -> demangle(addressToSymbol(x)), trace)), '\n') as trace "
                "FROM clusterAllReplicas('test_cluster_database_replicated', 'system.stack_trace') "
                "ORDER BY host, thread_id format Vertical\"".format(client), shell=True, stderr=subprocess.STDOUT).decode('utf-8')

        return subprocess.check_output("{} --allow_introspection_functions=1 --query "
               "\"SELECT arrayStringConcat(arrayMap(x, y -> concat(x, ': ', y), arrayMap(x -> addressToLine(x), trace), "
               "arrayMap(x -> demangle(addressToSymbol(x)), trace)), '\n') as trace "
               "FROM system.stack_trace format Vertical\"".format(client), shell=True, stderr=subprocess.STDOUT).decode('utf-8')
    except Exception as ex:
        print("Error occured while receiving stack traces from client: {}".format(str(ex)))
        return None

def get_server_pid(server_tcp_port):
    # lsof does not work in stress tests for some reason
    cmd_lsof = "lsof -i tcp:{port} -s tcp:LISTEN -Fp | awk '/^p[0-9]+$/{{print substr($0, 2)}}'".format(port=server_tcp_port)
    cmd_pidof = "pidof -s clickhouse-server"
    commands = [cmd_lsof, cmd_pidof]
    output = None
    for cmd in commands:
        try:
            output = subprocess.check_output(cmd, shell=True, stderr=subprocess.STDOUT, universal_newlines=True)
            if output:
                return int(output)
        except Exception as e:
            print("Cannot get server pid with {}, got {}: {}".format(cmd, output, e))
    return None # most likely server dead

def is_selected_suite(selected_suite_names, suite):
    if selected_suite_names == None:
        return True
    for selected_suite_name in selected_suite_names:
        if selected_suite_name in suite:
            return True
    return False

def colored(text, args, color=None, on_color=None, attrs=None):
    if termcolor and (sys.stdout.isatty() or args.force_color):
        return termcolor.colored(text, color, on_color, attrs)
    else:
        return text


stop_time = None
exit_code = multiprocessing.Value("i", 0)
server_died = multiprocessing.Event()
stop_tests_triggered_lock = multiprocessing.Lock()
stop_tests_triggered = multiprocessing.Event()
queue = multiprocessing.Queue(maxsize=1)
restarted_tests = []  # (test, stderr)

# def run_tests_array(all_tests, suite, suite_dir, suite_tmp_dir, run_total):
def run_tests_array(all_tests_with_params):
    all_tests, num_tests, suite, suite_dir, suite_tmp_dir = all_tests_with_params
    global stop_time
    global exit_code
    global server_died

    OP_SQUARE_BRACKET = colored("[", args, attrs=['bold'])
    CL_SQUARE_BRACKET = colored("]", args, attrs=['bold'])

    MSG_FAIL = OP_SQUARE_BRACKET + colored(" FAIL ", args, "red", attrs=['bold']) + CL_SQUARE_BRACKET
    MSG_UNKNOWN = OP_SQUARE_BRACKET + colored(" UNKNOWN ", args, "yellow", attrs=['bold']) + CL_SQUARE_BRACKET
    MSG_OK = OP_SQUARE_BRACKET + colored(" OK ", args, "green", attrs=['bold']) + CL_SQUARE_BRACKET
    MSG_SKIPPED = OP_SQUARE_BRACKET + colored(" SKIPPED ", args, "cyan", attrs=['bold']) + CL_SQUARE_BRACKET

    passed_total = 0
    skipped_total = 0
    failures_total = 0
    failures = 0
    failures_chain = 0
    start_time = datetime.now()

    is_concurrent = multiprocessing.current_process().name != "MainProcess"

    client_options = get_additional_client_options(args)

    def print_test_time(test_time):
        if args.print_time:
            return " {0:.2f} sec.".format(test_time)
        else:
            return ''

    def judge_case_fail(proc, stdout,stderr,total_time,status,testcase_args, stdout_file, stderr_file, reference_file):
        failures = 0
        failures_chain = 0
        passed_total = 0
        OP_SQUARE_BRACKET = colored("[", args, attrs=['bold'])
        CL_SQUARE_BRACKET = colored("]", args, attrs=['bold'])
        MSG_FAIL = OP_SQUARE_BRACKET + colored(" FAIL ", args, "red", attrs=['bold']) + CL_SQUARE_BRACKET
        MSG_UNKNOWN = OP_SQUARE_BRACKET + colored(" UNKNOWN ", args, "yellow", attrs=['bold']) + CL_SQUARE_BRACKET
        MSG_OK = OP_SQUARE_BRACKET + colored(" OK ", args, "green", attrs=['bold']) + CL_SQUARE_BRACKET
        if proc.returncode is None:
            try:
                proc.kill()
            except OSError as e:
                if e.errno != ESRCH:
                    raise

            failures = 1
            status += MSG_FAIL
            status += print_test_time(total_time)
            status += " - Timeout!\n"
            if stderr:
                status += stderr
            status += 'Database: ' + testcase_args.testcase_database if type(testcase_args) is not tuple else 'failed to create'
        else:
            if proc.returncode != 0:
                failures = 1
                failures_chain = 1
                status += MSG_FAIL
                status += print_test_time(total_time)
                status += ' - return code {}\n'.format(proc.returncode)

                if stderr:
                    status += stderr

                # Stop on fatal errors like segmentation fault. They are sent to client via logs.
                if ' <Fatal> ' in stderr:
                    server_died.set()

                if testcase_args.stop and ('Connection refused' in stderr or 'Attempt to read after eof' in stderr) and not 'Received exception from server' in stderr:
                    server_died.set()

                if os.path.isfile(stdout_file):
                    status += ", result:\n\n"
                    status += '\n'.join(
                        open(stdout_file).read().split('\n')[:100])
                    status += '\n'

                status += 'Database: ' + testcase_args.testcase_database if type(testcase_args) is not tuple else 'failed to create'

            elif stderr and stderr != '\n':
                failures = 1
                failures_chain = 1
                status += MSG_FAIL
                status += print_test_time(total_time)
                status += " - having stderror:\n{}\n".format(
                    '\n'.join(stderr.split('\n')[:100]))
                status += 'Database: ' + testcase_args.testcase_database
            elif 'Exception' in stdout:
                failures = 1
                failures_chain = 1
                status += MSG_FAIL
                status += print_test_time(total_time)
                status += " - having exception:\n{}\n".format(
                    '\n'.join(stdout.split('\n')[:100]))
                status += 'Database: ' + testcase_args.testcase_database
            elif not os.path.isfile(reference_file):
                status += MSG_UNKNOWN
                status += print_test_time(total_time)
                status += " - no reference file\n"
                status += 'Database: ' + testcase_args.testcase_database
            else:
                result_is_different = subprocess.call(['diff', '-q', reference_file, stdout_file], stdout=PIPE)

                if result_is_different:
                    diff = Popen(['diff', '-U', str(testcase_args.unified), reference_file, stdout_file], stdout=PIPE, universal_newlines=True, errors='replace').communicate()[0]
                    failures = 1
                    status += MSG_FAIL
                    status += print_test_time(total_time)
                    status += " - result differs with reference:\n{}\n".format(diff)
                    status += 'Database: ' + testcase_args.testcase_database
                else:
                    if testcase_args.test_runs > 1 and total_time > 60 and 'long' not in name:
                        # We're in Flaky Check mode, check the run time as well while we're at it.
                        failures = 1
                        failures_chain = 1
                        status += MSG_FAIL
                        status += print_test_time(total_time)
                        status += " - Test runs too long (> 60s). Make it faster.\n"
                        status += 'Database: ' + testcase_args.testcase_database
                    else:
                        passed_total = 1
                        failures_chain = 0
                        status += MSG_OK
                        status += print_test_time(total_time)
                        status += "\n"
                        if os.path.exists(stdout_file):
                            os.remove(stdout_file)
                        if os.path.exists(stderr_file):
                            os.remove(stderr_file)
        return failures,failures_chain,passed_total,status

    if num_tests > 0:
        about = 'about ' if is_concurrent else ''
        print(f"\nRunning {about}{num_tests} {suite} tests ({multiprocessing.current_process().name}).\n")

    while True:
        if is_concurrent:
            case = queue.get()
            if not case:
                break
        else:
            if all_tests:
                case = all_tests.pop(0)
            else:
                break

        if server_died.is_set():
            stop_tests()
            break

        if stop_time and time() > stop_time:
            print("\nStop tests run because global time limit is exceeded.\n")
            stop_tests()
            break

        case_file = os.path.join(suite_dir, case)
        (name, ext) = os.path.splitext(case)

        try:
            status = ''
            if not is_concurrent:
                sys.stdout.flush()
                sys.stdout.write("{0:72}".format(name + ": "))
                # This flush is needed so you can see the test name of the long
                # running test before it will finish. But don't do it in parallel
                # mode, so that the lines don't mix.
                sys.stdout.flush()
            else:
                status = "{0:72}".format(name + ": ")

            if args.skip and any(s in name for s in args.skip):
                status += MSG_SKIPPED + " - skip\n"
                skipped_total += 1
            elif not args.zookeeper and ('zookeeper' in name
                    or 'replica' in name):
                status += MSG_SKIPPED + " - no zookeeper\n"
                skipped_total += 1
            elif not args.shard and ('shard' in name
                    or 'distributed' in name
                    or 'global' in name):
                status += MSG_SKIPPED + " - no shard\n"
                skipped_total += 1
            elif not args.no_long and ('long' in name
                    # Tests for races and deadlocks usually are runned in loop
                    #  for significant amount of time
                    or 'deadlock' in name
                    or 'race' in name):
                status += MSG_SKIPPED + " - no long\n"
                skipped_total += 1
            else:
                disabled_file = os.path.join(suite_dir, name) + '.disabled'

                if os.path.exists(disabled_file) and not args.disabled:
                    message = open(disabled_file, 'r').read()
                    status += MSG_SKIPPED + " - " + message + "\n"
                else:

                    if args.testname:
                        clickhouse_proc = Popen(shlex.split(args.client), stdin=PIPE, stdout=PIPE, stderr=PIPE, universal_newlines=True)
                        failed_to_check = False
                        try:
                            clickhouse_proc.communicate(("SELECT 'Running test {suite}/{case} from pid={pid}';".format(pid = os.getpid(), case = case, suite = suite)), timeout=20)
                        except:
                            failed_to_check = True

                        if failed_to_check or clickhouse_proc.returncode != 0:
                            failures += 1
                            print("Server does not respond to health check")
                            server_died.set()
                            stop_tests()
                            break

                    file_suffix = ('.' + str(os.getpid())) if is_concurrent and args.test_runs > 1 else ''
                    reference_file = os.path.join(suite_dir, name) + '.reference'
                    stdout_file = os.path.join(suite_tmp_dir, name) + file_suffix + '.stdout'
                    stderr_file = os.path.join(suite_tmp_dir, name) + file_suffix + '.stderr'

                    testcase_args = configure_testcase_args(args, case_file, suite_tmp_dir, stderr_file)
                    if type(testcase_args) is tuple:
                        proc, stdout, stderr, total_time = testcase_args
                    else:
                        proc, stdout, stderr, total_time = run_single_test(testcase_args, ext, server_logs_level, client_options, case_file, stdout_file, stderr_file)
                    flag_failed_retry = 1
                    temp_failures = 0
                    temp_failures_chain = 0
                    while flag_failed_retry and flag_failed_retry <= MAX_RETRIES:
                        temp_failures,temp_failures_chain,passed_total,status = judge_case_fail(proc, stdout, stderr, total_time, status, testcase_args, stdout_file, stderr_file,reference_file)
                        if temp_failures:
                            if flag_failed_retry < MAX_RETRIES:
                                status = status.replace('[ FAIL ]',f'[ fail_retry:{str(flag_failed_retry)} ]')
                                if status and not status.endswith('\n'):
                                    status += '\n'
                                restarted_tests.append((case_file, stderr))
                                testcase_args = configure_testcase_args(args, case_file, suite_tmp_dir, stderr_file)
                                if type(testcase_args) is tuple:
                                    proc, stdout, stderr, total_time = testcase_args
                                else:
                                    proc, stdout, stderr, total_time = run_single_test(testcase_args, ext, server_logs_level, client_options, case_file, stdout_file, stderr_file)

                                status += "{0:72}".format(name + ": ")
                                flag_failed_retry += 1
                                sleep(2 ** flag_failed_retry)
                            else:
                                break
                        else:
                            break
                    if MAX_RETRIES < flag_failed_retry:
                        if args.replicated_database:
                            if DISTRIBUTED_DDL_TIMEOUT_MSG in stderr:
                                server_died.set()
                    failures = failures + temp_failures
                    failures_chain = failures_chain + temp_failures_chain

            if status and not status.endswith('\n'):
                status += '\n'

            sys.stdout.write(status)
            sys.stdout.flush()

            if args.locate_asan_error_to_case:
                sleep(0.2)
                collect_new_asan_log_after_test(case_name=case)

        except KeyboardInterrupt as e:
            print(colored("Break tests execution", args, "red"))
            stop_tests()
            raise e
        except:
            exc_type, exc_value, tb = sys.exc_info()
            failures += 1
            fail_case_name = "{0:72}".format(name + ": ")
            print("{0} {1} - Test internal error: {2}\n{3}\n{4}".format(fail_case_name, MSG_FAIL, exc_type.__name__,exc_value,"\n".join(traceback.format_tb(tb, 10))))

        if failures_chain >= 30:
            stop_tests()
            break

    failures_total = failures_total + failures

    if failures_total > 0:
        print(colored(f"\nHaving {failures_total} errors! {passed_total} tests passed."
            f" {skipped_total} tests skipped. {(datetime.now() - start_time).total_seconds():.2f} s elapsed"
            f' ({multiprocessing.current_process().name}).',
            args, "red", attrs=["bold"]))
        exit_code.value = 1
    else:
        print(colored(f"\n{passed_total} tests passed. {skipped_total} tests skipped."
            f" {(datetime.now() - start_time).total_seconds():.2f} s elapsed"
            f' ({multiprocessing.current_process().name}).',
            args, "green", attrs=["bold"]))

    sys.stdout.flush()


server_logs_level = "warning"


def check_server_started(client, retry_count):
    print("Connecting to ClickHouse server...", end='')
    sys.stdout.flush()
    while retry_count > 0:
        clickhouse_proc = Popen(shlex.split(client), stdin=PIPE, stdout=PIPE, stderr=PIPE)
        (stdout, stderr) = clickhouse_proc.communicate(b"SELECT 1")

        if clickhouse_proc.returncode == 0 and stdout.startswith(b"1"):
            print(" OK")
            sys.stdout.flush()
            return True

        if clickhouse_proc.returncode == 210:
            # Connection refused, retry
            print('.', end = '')
            sys.stdout.flush()
            retry_count -= 1
            sleep(0.5)
            continue

        # Other kind of error, fail.
        print('')
        print("Client invocation failed with code ", clickhouse_proc.returncode, ": ")
        # We can't print this, because for some reason this is python 2,
        # and args appeared in 3.3. To hell with it.
        # print(''.join(clickhouse_proc.args))
        print("stdout: ")
        print(stdout)
        print("stderr: ")
        print(stderr)
        sys.stdout.flush()
        return False

    print('')
    print('All connection tries failed')
    sys.stdout.flush()

    return False


class BuildFlags():
    THREAD = 'thread-sanitizer'
    ADDRESS = 'address-sanitizer'
    UNDEFINED = 'ub-sanitizer'
    MEMORY = 'memory-sanitizer'
    DEBUG = 'debug-build'
    UNBUNDLED = 'unbundled-build'
    RELEASE = 'release-build'
    DATABASE_ORDINARY = 'database-ordinary'
    POLYMORPHIC_PARTS = 'polymorphic-parts'
    DATABASE_REPLICATED = 'database-replicated'
    S3 = 's3'
    MULTI_SERVERS = 'multi-servers'


def collect_build_flags(client):
    clickhouse_proc = Popen(shlex.split(client), stdin=PIPE, stdout=PIPE, stderr=PIPE)
    (stdout, stderr) = clickhouse_proc.communicate(b"SELECT value FROM system.build_options WHERE name = 'CXX_FLAGS'")
    result = []

    if clickhouse_proc.returncode == 0:
        if b'-fsanitize=thread' in stdout:
            result.append(BuildFlags.THREAD)
        elif b'-fsanitize=address' in stdout:
            result.append(BuildFlags.ADDRESS)
        elif b'-fsanitize=undefined' in stdout:
            result.append(BuildFlags.UNDEFINED)
        elif b'-fsanitize=memory' in stdout:
            result.append(BuildFlags.MEMORY)
    else:
        raise Exception("Cannot get information about build from server errorcode {}, stderr {}".format(clickhouse_proc.returncode, stderr))

    clickhouse_proc = Popen(shlex.split(client), stdin=PIPE, stdout=PIPE, stderr=PIPE)
    (stdout, stderr) = clickhouse_proc.communicate(b"SELECT value FROM system.build_options WHERE name = 'BUILD_TYPE'")

    if clickhouse_proc.returncode == 0:
        if b'Debug' in stdout:
            result.append(BuildFlags.DEBUG)
        elif b'RelWithDebInfo' in stdout or b'Release' in stdout:
            result.append(BuildFlags.RELEASE)
    else:
        raise Exception("Cannot get information about build from server errorcode {}, stderr {}".format(clickhouse_proc.returncode, stderr))

    clickhouse_proc = Popen(shlex.split(client), stdin=PIPE, stdout=PIPE, stderr=PIPE)
    (stdout, stderr) = clickhouse_proc.communicate(b"SELECT value FROM system.build_options WHERE name = 'UNBUNDLED'")

    if clickhouse_proc.returncode == 0:
        if b'ON' in stdout or b'1' in stdout:
            result.append(BuildFlags.UNBUNDLED)
    else:
        raise Exception("Cannot get information about build from server errorcode {}, stderr {}".format(clickhouse_proc.returncode, stderr))

    clickhouse_proc = Popen(shlex.split(client), stdin=PIPE, stdout=PIPE, stderr=PIPE)
    (stdout, stderr) = clickhouse_proc.communicate(b"SELECT value FROM system.settings WHERE name = 'default_database_engine'")

    if clickhouse_proc.returncode == 0:
        if b'Ordinary' in stdout:
            result.append(BuildFlags.DATABASE_ORDINARY)
    else:
        raise Exception("Cannot get information about build from server errorcode {}, stderr {}".format(clickhouse_proc.returncode, stderr))

    clickhouse_proc = Popen(shlex.split(client), stdin=PIPE, stdout=PIPE, stderr=PIPE)
    (stdout, stderr) = clickhouse_proc.communicate(b"SELECT value FROM system.merge_tree_settings WHERE name = 'min_bytes_for_wide_part'")

    if clickhouse_proc.returncode == 0:
        if stdout == b'0\n':
            result.append(BuildFlags.POLYMORPHIC_PARTS)
    else:
        raise Exception("Cannot get inforamtion about build from server errorcode {}, stderr {}".format(clickhouse_proc.returncode, stderr))

    return result


def do_run_tests(jobs, suite, suite_dir, suite_tmp_dir, all_tests, parallel_tests, sequential_tests, parallel):
    if jobs > 1 and len(parallel_tests) > 0:
        print("Found", len(parallel_tests), "parallel tests and", len(sequential_tests), "sequential tests")
        run_n, run_total = parallel.split('/')
        run_n = float(run_n)
        run_total = float(run_total)
        tests_n = len(parallel_tests)
        if run_total > tests_n:
            run_total = tests_n

        if jobs > tests_n:
            jobs = tests_n
        if jobs > run_total:
            run_total = jobs

        batch_size = max(1, len(parallel_tests) // jobs)
        parallel_tests_array = []
        for _ in range(jobs):
            parallel_tests_array.append((None, batch_size, suite, suite_dir, suite_tmp_dir))

        with closing(multiprocessing.Pool(processes=jobs)) as pool:
            pool.map_async(run_tests_array, parallel_tests_array)

            for suit in parallel_tests:
                queue.put(suit)

            for _ in range(jobs):
                queue.put(None)

            queue.close()

        pool.join()

        run_tests_array((sequential_tests, len(sequential_tests), suite, suite_dir, suite_tmp_dir))
        return len(sequential_tests) + len(parallel_tests)
    else:
        num_tests = len(all_tests)
        run_tests_array((all_tests, num_tests, suite, suite_dir, suite_tmp_dir))
        return num_tests


def main(args):
    global server_died
    global stop_time
    global exit_code
    global server_logs_level

    def is_data_present():
        clickhouse_proc = Popen(shlex.split(args.client), stdin=PIPE, stdout=PIPE, stderr=PIPE)
        (stdout, stderr) = clickhouse_proc.communicate(b"EXISTS TABLE test.hits")
        if clickhouse_proc.returncode != 0:
            raise CalledProcessError(clickhouse_proc.returncode, args.client, stderr)

        return stdout.startswith(b'1')

    if not check_server_started(args.client, args.server_check_retries):
        raise Exception(
            "Server is not responding. Cannot execute 'SELECT 1' query. \
            Note: if you are using split build, you may have to specify -c option.")

    build_flags = collect_build_flags(args.client)
    if args.replicated_database:
        build_flags.append(BuildFlags.DATABASE_REPLICATED)

    if args.s3:
        build_flags.append(BuildFlags.S3)

    if not args.single_server:
        build_flags.append(BuildFlags.MULTI_SERVERS)

    if args.use_skip_list:
        tests_to_skip_from_list = collect_tests_to_skip(args.skip_list_path, build_flags)
    else:
        tests_to_skip_from_list = set([])

    if args.skip:
        args.skip = set(args.skip) | tests_to_skip_from_list
    else:
        args.skip = tests_to_skip_from_list

    if args.use_skip_list and not args.sequential:
        args.sequential = collect_sequential_list(args.skip_list_path)

    base_dir = os.path.abspath(args.queries)
    tmp_dir = os.path.abspath(args.tmp)

    # Keep same default values as in queries/shell_config.sh
    os.environ.setdefault("CLICKHOUSE_BINARY", args.binary)
    #os.environ.setdefault("CLICKHOUSE_CLIENT", args.client)
    os.environ.setdefault("CLICKHOUSE_CONFIG", args.configserver)
    if args.configclient:
        os.environ.setdefault("CLICKHOUSE_CONFIG_CLIENT", args.configclient)

    # Force to print server warnings in stderr
    # Shell scripts could change logging level
    os.environ.setdefault("CLICKHOUSE_CLIENT_SERVER_LOGS_LEVEL", server_logs_level)

    # This code is bad as the time is not monotonic
    if args.global_time_limit:
        stop_time = time() + args.global_time_limit

    if args.zookeeper is None:
        _, out = subprocess.getstatusoutput(args.extract_from_config + " --try --config " + args.configserver + ' --key zookeeper | grep . | wc -l')
        try:
            if int(out) > 0:
                args.zookeeper = True
            else:
                args.zookeeper = False
        except ValueError:
            args.zookeeper = False

    if args.shard is None:
        _, out = subprocess.getstatusoutput(args.extract_from_config + " --try --config " + args.configserver + ' --key listen_host | grep -E "127.0.0.2|::"')
        if out:
            args.shard = True
        else:
            args.shard = False

    def create_common_database(args, db_name):
        #tenant_id.db_name is the name of the tenant database
        if args.tenant_id is None or len(args.tenant_id) == 0:
            create_database = db_name
        else:
            create_database = args.tenant_id + "." + db_name

        create_database_retries = 0
        while create_database_retries < MAX_RETRIES:
            clickhouse_proc_create = Popen(shlex.split(args.client), stdin=PIPE, stdout=PIPE, stderr=PIPE, universal_newlines=True)
            (_, stderr) = clickhouse_proc_create.communicate(("CREATE DATABASE IF NOT EXISTS " + db_name + get_db_engine(args, db_name)))
            if not need_retry(stderr):
                break
            create_database_retries += 1

        if args.tenant_id:
            create_database_retries = 0
            while create_database_retries < MAX_RETRIES:
                clickhouse_proc_create = Popen(shlex.split(args.client), stdin=PIPE, stdout=PIPE, stderr=PIPE, universal_newlines=True)
                (_, stderr) = clickhouse_proc_create.communicate(("CREATE DATABASE IF NOT EXISTS `" + create_database + "`"+ get_db_engine(args, db_name)))
                if not need_retry(stderr):
                    break
                create_database_retries += 1

    if args.database and args.database != "test":
        create_common_database(args, args.database)
    create_common_database(args, "test")

    def is_test_from_dir(suite_dir, case):
        case_file = os.path.join(suite_dir, case)
        (_, ext) = os.path.splitext(case)
        # We could also test for executable files (os.access(case_file, os.X_OK),
        # but it interferes with 01610_client_spawn_editor.editor, which is invoked
        # as a query editor in the test, and must be marked as executable.
        return os.path.isfile(case_file) and (ext in ['.sql', '.sh', '.py', '.expect'])

    def sute_key_func(item):
        if args.order == 'random':
            return random.random()

        if -1 == item.find('_'):
            return 99998, ''

        prefix, suffix = item.split('_', 1)

        try:
            return int(prefix), suffix
        except ValueError:
            return 99997, ''

    total_tests_run = 0
    for suite in sorted(os.listdir(base_dir), key=sute_key_func):
        if server_died.is_set():
            break

        suite_dir = os.path.join(base_dir, suite)
        suite_re_obj = re.search('^[0-9]+_(.*)$', suite)
        if not suite_re_obj: #skip .gitignore and so on
            continue

        suite_tmp_dir = os.path.join(tmp_dir, suite)
        if not os.path.exists(suite_tmp_dir):
            os.makedirs(suite_tmp_dir)

        suite = suite_re_obj.group(1)
        if os.path.isdir(suite_dir):

            if 'stateful' in suite and not args.no_stateful and not is_data_present():
                print("Won't run stateful tests because test data wasn't loaded.")
                continue
            if 'stateless' in suite and args.no_stateless:
                print("Won't run stateless tests because they were manually disabled.")
                continue
            if 'stateful' in suite and args.no_stateful:
                print("Won't run stateful tests because they were manually disabled.")
                continue
            if is_selected_suite(args.run, suite) == False:
                continue

            # Reverse sort order: we want run newest test first.
            # And not reverse subtests
            def key_func(item):
                if args.order == 'random':
                    return random.random()

                reverse = 1 if args.order == 'asc' else -1

                if -1 == item.find('_'):
                    return 99998

                prefix, _ = item.split('_', 1)

                try:
                    return reverse * int(prefix)
                except ValueError:
                    return 99997

            all_tests = os.listdir(suite_dir)
            all_tests = [case for case in all_tests if is_test_from_dir(suite_dir, case)]
            if args.test:
                all_tests = [t for t in all_tests if any(re.search(r, t) for r in args.test)]
            all_tests = all_tests * args.test_runs
            all_tests.sort(key=key_func)

            jobs = args.jobs
            parallel_tests = []
            sequential_tests = []
            for test in all_tests:
                if any(s in test for s in args.sequential):
                    sequential_tests.append(test)
                else:
                    parallel_tests.append(test)

            total_tests_run += do_run_tests(
                jobs, suite, suite_dir, suite_tmp_dir, all_tests, parallel_tests, sequential_tests, args.parallel)

    if args.hung_check:

        # Some queries may execute in background for some time after test was finished. This is normal.
        for _ in range(1, 60):
            timeout, processlist = get_processlist(args)
            if timeout or not processlist:
                break
            sleep(1)

        if timeout or processlist:
            if processlist:
                print(colored("\nFound hung queries in processlist:", args, "red", attrs=["bold"]))
                print(processlist)
            else:
                print(colored("Seems like server hung and cannot respond to queries", args, "red", attrs=["bold"]))

            clickhouse_tcp_port = os.getenv("CLICKHOUSE_PORT_TCP", '9000')
            server_pid = get_server_pid(clickhouse_tcp_port)
            bt = None
            if server_pid and not args.replicated_database:
                print("\nLocated ClickHouse server process {} listening at TCP port {}".format(server_pid, clickhouse_tcp_port))
                print("\nCollecting stacktraces from all running threads with gdb:")
                bt = get_stacktraces_from_gdb(server_pid)
                if len(bt) < 1000:
                    print("Got suspiciously small stacktraces: ", bt)
                    bt = None
            if bt is None:
                print("\nCollecting stacktraces from system.stacktraces table:")
                bt = get_stacktraces_from_clickhouse(args.client, args.replicated_database)
            if bt is None:
                print(
                    colored(
                        "\nUnable to locate ClickHouse server process listening at TCP port {}. "
                        "It must have crashed or exited prematurely!".format(clickhouse_tcp_port),
                        args, "red", attrs=["bold"]))
            else:
                print(bt)

            exit_code.value = 1
        else:
            print(colored("\nNo queries hung.", args, "green", attrs=["bold"]))

    if len(restarted_tests) > 0:
        print("\nSome tests were restarted:\n")
        for (test_case, stderr) in restarted_tests:
            print(test_case)
            print(stderr)
            print("\n")

    if not is_sanitizer_pass():
        exit_code.value = 1
        print('asan check failed！please check asan report in Job artifacts')

    if total_tests_run == 0:
        print("No tests were run.")
        sys.exit(1)
    else:
        print("All tests have finished.")

    sys.exit(exit_code.value)


def find_binary(name):
    if os.path.exists(name) and os.access(name, os.X_OK):
        return True
    paths = os.environ.get("PATH").split(':')
    for path in paths:
        if os.access(os.path.join(path, name), os.X_OK):
            return True

    # maybe it wasn't in PATH
    if os.access(os.path.join('/usr/local/bin', name), os.X_OK):
        return True
    if os.access(os.path.join('/usr/bin', name), os.X_OK):
        return True
    return False


def get_additional_client_options(args):
    if args.client_option:
        return ' '.join('--' + option for option in args.client_option)

    return ''


def get_additional_client_options_url(args):
    if args.client_option:
        return '&'.join(args.client_option)

    return ''


def collect_tests_to_skip(skip_list_path, build_flags):
    result = set([])
    if not os.path.exists(skip_list_path):
        return result

    with open(skip_list_path, 'r') as skip_list_file:
        content = skip_list_file.read()
        # allows to have comments in skip_list.json
        skip_dict = json.loads(json_minify(content))
        for build_flag in build_flags:
            cur_result = set(skip_dict[build_flag])
            print("Use build flag: {}, skips tests: {}".format(build_flag, cur_result))
            result |= cur_result 

    if len(result) > 0:
        print("Found file with skip-list {}, {} test will be skipped".format(skip_list_path, len(result)))

    return result


def collect_sequential_list(skip_list_path):
    if not os.path.exists(skip_list_path):
        return set([])

    with open(skip_list_path, 'r') as skip_list_file:
        content = skip_list_file.read()
        # allows to have comments in skip_list.json
        skip_dict = json.loads(json_minify(content))
        if 'parallel' in skip_dict:
            return skip_dict['parallel']
        return set([])


if __name__ == '__main__':
    # Move to a new process group and kill it at exit so that we don't have any
    # infinite tests processes left
    # (new process group is required to avoid killing some parent processes)
    os.setpgid(0, 0)
    signal.signal(signal.SIGTERM, signal_handler)
    signal.signal(signal.SIGINT, signal_handler)
    signal.signal(signal.SIGHUP, signal_handler)

    parser=ArgumentParser(description='ClickHouse functional tests')
    parser.add_argument('-q', '--queries', help='Path to queries dir')
    parser.add_argument('--tmp', help='Path to tmp dir')

    parser.add_argument('-b', '--binary', default='clickhouse',
        help='Path to clickhouse (if monolithic build, clickhouse-server otherwise) binary or name of binary in PATH')

    parser.add_argument('-c', '--client',
        help='Path to clickhouse-client (if split build, useless otherwise) binary of name of binary in PATH')

    parser.add_argument('--extract_from_config', help='extract-from-config program')
    parser.add_argument('--configclient', help='Client config (if you use not default ports)')
    parser.add_argument('--configserver', default= '/etc/clickhouse-server/config.xml', help='Preprocessed server config')
    parser.add_argument('-o', '--output', help='Output xUnit compliant test report directory')
    parser.add_argument('-t', '--timeout', type=int, default=600, help='Timeout for each test case in seconds')
    parser.add_argument('--global_time_limit', type=int, help='Stop if executing more than specified time (after current test finished)')
    parser.add_argument('test', nargs='*', help='Optional test case name regex')
    parser.add_argument('-d', '--disabled', action='store_true', default=False, help='Also run disabled tests')
    parser.add_argument('--stop', action='store_true', default=None, dest='stop', help='Stop on network errors')
    parser.add_argument('--order', default='desc', choices=['asc', 'desc', 'random'], help='Run order')
    parser.add_argument('--testname', action='store_true', default=None, dest='testname', help='Make query with test name before test run')
    parser.add_argument('--hung-check', action='store_true', default=False)
    parser.add_argument('--force-color', action='store_true', default=False)
    parser.add_argument('--database', help='Database for tests (random name test_XXXXXX by default)')
    parser.add_argument('--no-drop-if-fail', action='store_true', help='Do not drop database for test if test has failed')
    parser.add_argument('--hide-db-name', action='store_true', help='Replace random database name with "default" in stderr')
    parser.add_argument('--parallel', default='1/1', help='One parallel test run number/total')
    parser.add_argument('-j', '--jobs', default=1, nargs='?', type=int, help='Run all tests in parallel')
    parser.add_argument('--test-runs', default=1, nargs='?', type=int, help='Run each test many times (useful for e.g. flaky check)')
    parser.add_argument('-U', '--unified', default=3, type=int, help='output NUM lines of unified context')
    parser.add_argument('-r', '--server-check-retries', default=30, type=int, help='Num of tries to execute SELECT 1 before tests started')
    parser.add_argument('--skip-list-path', help="Path to skip-list file")
    parser.add_argument('--use-skip-list', action='store_true', default=False, help="Use skip list to skip tests if found")
    parser.add_argument('--db-engine', help='Database engine name')
    parser.add_argument('--replicated-database', action='store_true', default=False, help='Run tests with Replicated database engine')
    parser.add_argument('--no-stateless', action='store_true', help='Disable all stateless tests')
    parser.add_argument('--no-stateful', action='store_true', help='Disable all stateful tests')
    parser.add_argument('--skip', nargs='+', help="Skip these tests")
    parser.add_argument('--sequential', nargs='+', help="Run these tests sequentially even if --parallel specified")
    parser.add_argument('--no-long', action='store_false', dest='no_long', help='Do not run long tests')
    parser.add_argument('--client-option', nargs='+', help='Specify additional client argument')
    parser.add_argument('--print-time', action='store_true', dest='print_time', help='Print test time')
    group=parser.add_mutually_exclusive_group(required=False)
    group.add_argument('--zookeeper', action='store_true', default=None, dest='zookeeper', help='Run zookeeper related tests')
    group.add_argument('--no-zookeeper', action='store_false', default=None, dest='zookeeper', help='Do not run zookeeper related tests')
    group=parser.add_mutually_exclusive_group(required=False)
    group.add_argument('--shard', action='store_true', default=None, dest='shard', help='Run sharding related tests (required to clickhouse-server listen 127.0.0.2 127.0.0.3)')
    group.add_argument('--no-shard', action='store_false', default=None, dest='shard', help='Do not run shard related tests')
    parser.add_argument('--run', required=False, default=None, nargs='+', help='name of suites to be executed')
    parser.add_argument('--locate-asan-error-to-case', action='store_true', default=False,help='enable check asan erorr after each case finishs, no -j')
    parser.add_argument('--s3', required=False, action='store_true', help='Only run tests for S3 environment')
    parser.add_argument('--single-server', required=False, action='store_true', help='Add some tests which will fail on multi-server')
    parser.add_argument('--tenant_id', default='', help='set the tenant id for test')
    parser.add_argument('--disable_charset', default=True, help='disable character set for mysql pipeline')

    args = parser.parse_args()

    if args.queries and not os.path.isdir(args.queries):
        print("Cannot access the specified directory with queries (" + args.queries + ")", file=sys.stderr)
        sys.exit(1)

    # Autodetect the directory with queries if not specified
    if args.queries is None:
        args.queries = 'queries'

    if not os.path.isdir(args.queries):
        # If we're running from the repo
        args.queries = os.path.join(os.path.dirname(os.path.abspath( __file__ )), 'queries')

    if not os.path.isdir(args.queries):
        # Next we're going to try some system directories, don't write 'stdout' files into them.
        if args.tmp is None:
            args.tmp = '/tmp/clickhouse-test'

        args.queries = '/usr/local/share/clickhouse-test/queries'

    if not os.path.isdir(args.queries):
        args.queries = '/usr/share/clickhouse-test/queries'

    if not os.path.isdir(args.queries):
        print("Failed to detect path to the queries directory. Please specify it with '--queries' option.", file=sys.stderr)
        sys.exit(1)

    print("Using queries from '" + args.queries + "' directory")

    if args.skip_list_path is None:
        args.skip_list_path = os.path.join(args.queries, 'skip_list.json')

    if args.sequential is None:
        args.sequential = set([])

    if args.tmp is None:
        args.tmp = args.queries
    if args.client is None:
        if find_binary(args.binary + '-client'):
            args.client = args.binary + '-client'

            print("Using " + args.client + " as client program (expecting split build)")
        elif find_binary(args.binary):
            args.client = args.binary + ' client'

            print("Using " + args.client + " as client program (expecting monolithic build)")
        else:
            print("No 'clickhouse' or 'clickhouse-client' client binary found", file=sys.stderr)
            parser.print_help()
            sys.exit(1)

        if args.configclient:
            args.client += ' --config-file=' + args.configclient
        if os.getenv("CLICKHOUSE_HOST"):
            args.client += ' --host=' + os.getenv("CLICKHOUSE_HOST")
        if os.getenv("CLICKHOUSE_PORT_TCP"):
            args.client += ' --port=' + os.getenv("CLICKHOUSE_PORT_TCP")
        if os.getenv("CLICKHOUSE_DATABASE"):
            if args.tenant_id is None or len(args.tenant_id) == 0:
                args.client += ' --database=' + os.getenv("CLICKHOUSE_DATABASE")
            else:
                args.client += ' --database="' + args.tenant_id + "\`" + os.getenv("CLICKHOUSE_DATABASE") + '"'

    if args.client_option:
        # Set options for client
        if 'CLICKHOUSE_CLIENT_OPT' in os.environ:
            os.environ['CLICKHOUSE_CLIENT_OPT'] += ' '
        else:
            os.environ['CLICKHOUSE_CLIENT_OPT'] = ''

        os.environ['CLICKHOUSE_ADDITIONAL_CLIENT_OPT'] = get_additional_client_options(args)

        os.environ['CLICKHOUSE_CLIENT_OPT'] += get_additional_client_options(args)

        # Set options for curl
        if 'CLICKHOUSE_URL_PARAMS' in os.environ:
            os.environ['CLICKHOUSE_URL_PARAMS'] += '&'
        else:
            os.environ['CLICKHOUSE_URL_PARAMS'] = ''

        # this may have to be trimmed in some cases
        os.environ['CLICKHOUSE_ADDITIONAL_CLIENT_URL_PARAMS'] = get_additional_client_options_url(args)

        os.environ['CLICKHOUSE_URL_PARAMS'] += get_additional_client_options_url(args)

    if args.extract_from_config is None:
        if os.access(args.binary + '-extract-from-config', os.X_OK):
            args.extract_from_config = args.binary + '-extract-from-config'
        else:
            args.extract_from_config = args.binary + ' extract-from-config'

    if args.jobs is None:
        args.jobs = multiprocessing.cpu_count()

    if "ASAN_OPTIONS" in os.environ:
        ASAN_OPTIONS = os.environ['ASAN_OPTIONS'].split(',')
        for item in ASAN_OPTIONS:
            if item.startswith('log_path='):
                ASAN_LOG_PATH = os.path.dirname(item.replace('log_path=', ''))
                os.environ["ASAN_LOG_PATH"] = ASAN_LOG_PATH
                print('\nASAN enbaled, the ASAN_LOG_PATH is:', ASAN_LOG_PATH)

                if args.locate_asan_error_to_case:  # create folder for collect_new_asan_log_after_test()
                    individual_sanitizer_log_path = os.path.join(ASAN_LOG_PATH, 'tmp', 'individual_sanitizer_log', )
                    if not os.path.exists(individual_sanitizer_log_path):
                        os.system(f" mkdir -p '{individual_sanitizer_log_path}' ")

                break

    print(args)
    main(args)
